package com.kvn.blade.core.sender;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.bytes.ByteArrayDecoder;
import io.netty.handler.codec.bytes.ByteArrayEncoder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author wzy
 * @date 2017年11月20日 上午11:07:37
 */
public class NettyClient extends ChannelInboundHandlerAdapter {
	private static final Logger logger = LoggerFactory.getLogger(NettyClient.class);

	private String host;
	private int port;

	private Object response;

	private final Object obj = new Object();

	public NettyClient(String host, int port) {
		super();
		this.host = host;
		this.port = port;
	}

	@Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
		this.response = msg;

		synchronized (obj) {
			obj.notifyAll(); // 收到响应，唤醒线程
		}
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
		logger.error("client caught exception", cause);
        ctx.close();
	}

	public Object send(Object request) throws Exception {
		EventLoopGroup group = new NioEventLoopGroup();
		try {
			Bootstrap bootstrap = new Bootstrap();
			bootstrap.group(group).channel(NioSocketChannel.class).option(ChannelOption.SO_KEEPALIVE, true).handler(new ChannelInitializer<SocketChannel>() {
				@Override
				public void initChannel(SocketChannel channel) throws Exception {
					channel.pipeline()
//					.addLast(new StringDecoder(CharsetUtil.UTF_8)) // 将 RPC 响应进行解码（为了处理响应）
//					.addLast(new StringEncoder(CharsetUtil.UTF_8)) // 将RPC请求进行编码（为了发送请求）
//                   为了通用性，使用 byte[] 来进行数据传输，具体到 DTO 的对象编解码做为使用者通用的业务逻辑层，放到 com.kvn.blade.core.decoder.Decoder 和
//                   com.kvn.blade.core.encoder.Encoder 中去扩展
					.addLast(new ByteArrayEncoder()) // 将RPC请求进行编码（为了发送请求）
					.addLast(new ByteArrayDecoder()) // 将RPC请求进行解码（为了处理响应）
							.addLast(NettyClient.this); // 使用 RpcClient 发送 RPC 请求
				}
			});

			ChannelFuture future = bootstrap.connect(host, port).sync();
			future.channel().writeAndFlush(request).sync();

			synchronized (obj) {
				obj.wait(); // 未收到响应，使线程等待
			}

			if (response != null) {
				future.channel().closeFuture().sync();
			}
			System.out.println("返回结果==>" + response);
			return response;
		} finally {
			group.shutdownGracefully();
		}
	}
	
	public static void main(String[] args) throws Exception {
		NettyClient client = new NettyClient("localhost", 8091);
		Object rlt = client.send("123\n\r");
		System.out.println("====>" + rlt);
	}

}
